#include <rte_ethdev.h>

#include "runtime/reactor.h"
#include "runtime/network_poller.h"
#include "util/logger.h"

using namespace kawauso;
using namespace std;

Reactor::Reactor(NetworkOptions &networkOptions, RuntimeOptions &runtimeOptions)
    : networkOptions(networkOptions)
    , runtimeOptions(runtimeOptions) {}

void Reactor::init() {
    initDPDKEnvironment();
    registerPollers();
}

void Reactor::initDPDKEnvironment() {

    vector<string> originArgs;
    vector<char *> cookedArgs;

    // set program's name
    originArgs.emplace_back("EAL1");
    // set the namespace for the file resources used by DPDK, providing support for running multiple gateway processes
    originArgs.emplace_back(fmt::format("--file-prefix={}", getpid()));

    for (auto &deviceInfo : networkOptions.directDevices) {
        originArgs.emplace_back(fmt::format("-a{}", deviceInfo.pciAddress));
    }

    for (auto &deviceInfo : networkOptions.emulateDevices) {
        originArgs.emplace_back(fmt::format(
                "--vdev=eth_af_packet_{0},iface={0},qpairs={1}",
                deviceInfo.attachInterface, runtimeOptions.coreThreadNum));
    }

    for (auto &arg : originArgs) {
        cookedArgs.push_back(arg.data());
    }

    int res = rte_eal_init((int)cookedArgs.size(), cookedArgs.data());
    if (res < 0) {
        throw std::runtime_error("Failed to setup DPDK environment abstraction layer");
    }
}

void Reactor::registerPollers() {
    pollers.push_back(new NetworkPoller(networkOptions));
}

void Reactor::start() {

    int res = rte_eal_mp_remote_launch([] (void *ptr) {
        try {
            ((Reactor *)ptr)->runEventLoop();
        } catch (std::exception &e) {
            error("{}", e.what());
            return EXIT_FAILURE;
        }
        return EXIT_SUCCESS;
    }, this, rte_rmt_call_main_t::CALL_MAIN);

    if (res < 0) {
        error("Failed to start event loop on all cores");
        stop();
    }

    rte_eal_mp_wait_lcore();
}

void Reactor::runEventLoop() {

    uint16_t emptyPollCount = 0;

    while (isRunning) {

        for (auto &poller : pollers) {
            if (!poller->poll()) {
                emptyPollCount ++;
            }
        }

        if (unlikely(emptyPollCount == pollers.size())) {
            if (emptyPollCount < runtimeOptions.suspendThreshold) {
                rte_delay_us_block(emptyPollCount);
            } else {
                uint16_t suspendTime = min(emptyPollCount, runtimeOptions.maxSuspendTime);
                rte_delay_us_sleep(suspendTime);
            }
        }
    }
}

Reactor::~Reactor() {
    int res = rte_eal_cleanup();
    if (res < 0) {
        error("Failed to cleanup DPDK environment abstraction layer");
    }
}

void Reactor::stop() {
    isRunning = false;
}
